package com.zyx.pulsardemo.protobuf;

import com.zyx.pulsardemo.protobuf.pojo.DogPOJO;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

import java.util.concurrent.TimeUnit;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/21 15:16
 * desc: 消费 Protobuf 信息
 */
@Slf4j
public class ConsumeProtobufDemo {

    private static volatile boolean stop = false;

    public static void main(String[] args) throws InterruptedException {
        ConsumeProtobufDemo demo = new ConsumeProtobufDemo();
        PulsarClient client = demo.getClient();
        log.info("client =====>>>>> {}", client);
        demo.startConsumer(client);
        Thread.sleep(30 * 1000);
        // 30s后终止消费线程
        stop = true;
        System.out.println("stop =====>>>>> true");
    }

    private final String serviceUrl;
    protected String topic;
    protected String subscriptionName;

    public ConsumeProtobufDemo() {
        this.serviceUrl = "pulsar://bigdata022:6650,bigdata023:6650,bigdata024:6650";
        this.topic = "persistent://manning/protobuf/example-topic";
        this.subscriptionName = "my-share-sub";
    }

    protected void startConsumer(PulsarClient client) {
        Consumer<DogPOJO.Dog> consumer = getConsumer(client);
        try {
            consumer.seek(MessageId.latest);
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
        Runnable run = () -> {
            log.info("prepate to consumer...");
            log.info("stop =====>>>>> {}", stop);
            while (true) {
                if(stop) {
                    log.info("begin to stop");
                    try {
                        consumer.close();
                        client.close();
                    } catch (Exception e) {
                        log.error("close fail, caused by:{}", e.getMessage());
                        e.printStackTrace();
                    }
                    break;
                }
                // Wait for a message
                try {
                    // 处理消息, receive会一条一条处理, 如果没有数据会一直
                    Messages<DogPOJO.Dog> messages = consumer.batchReceive();
                    if (messages != null && messages.size() > 0) {
                        for (Message<DogPOJO.Dog> message : messages) {
                            log.info("dog =====>>>>> {}", message.getValue());
                            consumer.acknowledgeAsync(message);
                        }
                    }
                    Thread.sleep(100);
                } catch (Exception e) {
                    log.error("Unable to consume message: {}", e.getMessage());
                }
            }
            log.info("stop the consumer thread...");
        };
        new Thread(run, "consumer-thread").start();
    }

    protected PulsarClient getClient() {
        try {
            return PulsarClient.builder().serviceUrl(serviceUrl).build();
        } catch (PulsarClientException e) {
            log.warn("get client fails, caused by: {}", e.getMessage());
            e.printStackTrace();
        }
        return null;
    }

    protected Consumer<DogPOJO.Dog> getConsumer(PulsarClient client) {
        try {
            return client
                    .newConsumer(Schema.PROTOBUF(DogPOJO.Dog.class))
                    .topic(topic)
                    .subscriptionName(subscriptionName)
                    .subscriptionType(SubscriptionType.Exclusive)
                    // 批量接收操作完成：足够数量的消息、消息字节数、等待超时
                    .batchReceivePolicy(BatchReceivePolicy.builder()
                            // 设置批量获取的消息数量, 100条
                            .maxNumMessages(100)
                            // 设置批量获取的消息字节数, 1MB
                            .maxNumBytes(1024 * 1024)
                            // 设置批量获取的等待超时, 200ms
                            .timeout(200, TimeUnit.MILLISECONDS)
                            .build())
                    .subscribe();
        } catch (PulsarClientException e) {
            log.warn("consumer creates fail, caused by: {}", e.getMessage());
            e.printStackTrace();
        }
        return null;
    }
}
